package com.seari.main;

import com.seari.bean.DaduhuiDataBean;
import com.seari.client.BaseClient;
import com.seari.client.kafka.KafkaStarter;
import com.seari.commonutils.redis.JedisUtilProxy;
import com.seari.controller.MetroDataController;
import com.seari.service.AppCacheService;
import com.seari.service.TrainService;
import com.seari.utils.MiscTools;
import com.seari.utils.SpringContainer;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.Properties;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

@Component
@Order(value = 1)
public class StartupClass implements ApplicationRunner
{
	protected static Logger logger = LoggerFactory.getLogger(StartupClass.class);
	
	public static Properties config; 
	
	static String configFilePath = "config.cfg";
	
	private ScheduledExecutorService executor = Executors.newScheduledThreadPool(10);
	
	@Autowired
	private AppCacheService appCacheService;
	
	@Autowired
	private TrainService trainService;
	
	@Autowired
	private KafkaStarter kafkaStarter;
	
	@Override
	public void run(ApplicationArguments args) throws Exception
	{
		System.out.println("start up class loading...");
	    JedisUtilProxy.initJedisPoolUtil("127.0.0.1", 6379);
	    init();
	}
	
	public void init()
	{
		//加载配置文件
		try
		{
			config = MiscTools.loadConfig(configFilePath);
			logger.info("read config file complete");
			//预热大都会缓存
			appCacheService.refreshCache(MetroDataController.METRO_FLOW_KEY, new HashMap<String, DaduhuiDataBean>());
			//初始化车次窗数据
			trainService.initTrainData();
			//初始化KAFKA处理任务
			if(Boolean.parseBoolean(config.getProperty("kafka_actived","true")))
			{
				kafkaStarter.init();
			}
			//初始化周期性任务
			executeTask();
		} catch (Exception e)
		{
			// TODO: handle exception
			logger.warn(e.getMessage());
		}
	}
	
	public void executeTask()
    {
    	String clientListStr = config.getProperty("client_list","");
    	if(StringUtils.isBlank(clientListStr)) 
    	{
    		//client列表为空时停止执行
    		executor.shutdown();
    		return;
    	} 
    	String[] clientList = clientListStr.split("\\|");
    	for(int i = 0; i < clientList.length; i++)
    	{
    		String cleintStr = clientList[i];
    		try
			{
				//BaseClient baseClient = (BaseClient) Class.forName("com.seari.client." + cleintStr).newInstance();
    			BaseClient baseClient = (BaseClient) SpringContainer.getBean(cleintStr);
				String timeGap = config.getProperty(cleintStr + "_timeGap");
				String url = config.getProperty(cleintStr + "_url");
				String firstRunDelay = config.getProperty(cleintStr + "_delay", "0");
				baseClient.setUrl(url);
				executor.scheduleWithFixedDelay(baseClient, Integer.parseInt(firstRunDelay) , Integer.parseInt(timeGap), TimeUnit.SECONDS);
			} catch (Exception e)
			{
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
    		
    	}
    }
}
